package mr

import (
	"encoding/json"
	"fmt"
	"hash/fnv"
	"io/ioutil"
	"log"
	"net/rpc"
	"os"
	"sort"
	"strconv"
	"time"
)

// Map functions return a slice of KeyValue.
type KeyValue struct {
	Key   string
	Value string
}

type ByKey []KeyValue

func (a ByKey) Len() int           { return len(a) }
func (a ByKey) Swap(i, j int)      { a[i], a[j] = a[j], a[i] }
func (a ByKey) Less(i, j int) bool { return a[i].Key < a[j].Key }

// use ihash(key) % NReduce to choose the reduce
// task number for each KeyValue emitted by Map.
func ihash(key string) int {
	h := fnv.New32a()
	h.Write([]byte(key))
	return int(h.Sum32() & 0x7fffffff)
}

// main/mrworker.go calls this function.
func Worker(mapf func(string, string) []KeyValue, reducef func(string, []string) string) {
	work := true
	for work {
		args := ExampleArgs{}
		task := Task{}
		call("Master.AllotTask", &args, &task)
		switch task.TaskType {
		case MapType:
			{
				DoMap(&task, mapf)

				taskDone(task.Id)
			}
		case ReduceType:
			{
				DoReduce(&task, reducef)

				taskDone(task.Id)

			}
		case WaitingType:
			{

				time.Sleep(time.Second)
			}
		case OverType:
			{

				work = false
			}
		}
	}
	// Your worker implementation here.

	// uncomment to send the Example RPC to the master.
	// CallExample()
}

// example function to show how to make an RPC call to the master.
// the RPC argument and reply types are defined in rpc.go.
func CallExample() {
	args := ExampleArgs{}
	task := Task{}
	call("Master.AllotTask", &args, &task)
}

// send an RPC request to the master, wait for the response.
// usually returns true.
// returns false if something goes wrong.
func call(rpcname string, args interface{}, reply interface{}) bool {
	// c, err := rpc.DialHTT：P(":tcp", "127.0.0.1"+":1234")
	sockname := masterSock()
	c, err := rpc.DialHTTP("unix", sockname)
	if err != nil {
		log.Fatal("dialing:", err)
	}
	defer c.Close()

	err = c.Call(rpcname, args, reply)
	if err == nil {
		return true
	}

	fmt.Println(err)
	return false
}

func taskDone(taskId int) {
	reply := &ExampleReply{}
	call("Master.TaskDone", taskId, reply)
}

func DoMap(task *Task, f func(string, string) []KeyValue) {
	intermediate := []KeyValue{}
	for _, filename := range task.Inputfiles {
		file, err := os.Open(filename)
		if err != nil {
			log.Fatalf("cannot open %v", filename)
		}
		content, err := ioutil.ReadAll(file)
		if err != nil {
			log.Fatalf("cannot read %v", filename)
		}
		file.Close()
		kva := f(filename, string(content))
		intermediate = append(intermediate, kva...)
	}
	//initialize and loop over []KeyValue
	sort.Sort(ByKey(intermediate))
	rn := task.ReduceNum
	HashedKV := make([][]KeyValue, rn)

	for _, kv := range intermediate {
		HashedKV[ihash(kv.Key)%rn] = append(HashedKV[ihash(kv.Key)%rn], kv)
	}
	// _, err := os.Stat("./tmp")
	// if err != nil {
	// 	os.Mkdir("./tmp", 0777)
	// }

	for i := 0; i < rn; i++ {
		oname := "mr-tmp-" + strconv.Itoa(task.Id) + "-" + strconv.Itoa(i)
		ofile, _ := os.Create(oname)
		enc := json.NewEncoder(ofile)
		for _, kv := range HashedKV[i] {
			enc.Encode(kv)
		}
		ofile.Close()
	}
}

func DoReduce(task *Task, reducef func(string, []string) string) {

	oname := "mr-out-" + strconv.Itoa(task.Id)
	ofile, _ := os.Create(oname)
	intermediate := readFromLocalFile(task.Inputfiles)
	sort.Sort(ByKey(intermediate))

	i := 0
	for i < len(intermediate) {
		j := i + 1
		for j < len(intermediate) && intermediate[j].Key == intermediate[i].Key {
			j++
		}
		values := []string{}
		for k := i; k < j; k++ {
			values = append(values, intermediate[k].Value)
		}
		output := reducef(intermediate[i].Key, values)
		fmt.Fprintf(ofile, "%v %v\n", intermediate[i].Key, output)
		i = j
	}
	ofile.Close()
}

func readFromLocalFile(files []string) []KeyValue {
	var kva []KeyValue
	for _, filepath := range files {
		file, _ := os.Open(filepath)
		dec := json.NewDecoder(file)
		for {
			var kv KeyValue
			if err := dec.Decode(&kv); err != nil {
				break
			}
			kva = append(kva, kv)
		}
		file.Close()
	}
	return kva
}
